package vip.meeet;

import com.alibaba.fastjson.JSONObject;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import vip.meeet.elasticsearch.bean.ScrollResponse;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;

public class MigrateHelper {

    public static void migrate(String hostPort,String fromIndex,String toIndex,String time) throws IOException {
        Optional<ScrollResponse> scrollResponseOptional = DocHelper.scrollSource(hostPort, fromIndex, time,20000);
        ScrollResponse scrollResponse;
        String lastScrollId = null;
        while (scrollResponseOptional.isPresent()) {
            scrollResponse = scrollResponseOptional.get();
            if(scrollResponse.getSources().size() == 0){
                break;
            }
            lastScrollId = scrollResponse.getScrollId();
            List<JSONObject> sources = scrollResponse.getSources();
            BulkBuilder builder = BulkBuilder.getBuilder();
            for (JSONObject source : sources) {
                String content = source.toJSONString();
                builder.index(toIndex,source.getString("_id"),content);
//                System.out.println(content);
            }
            builder.postBulk(hostPort);
            scrollResponseOptional = DocHelper.scrollSourceWithId(hostPort,lastScrollId,time);
        }
        //删除scrollId
        if(lastScrollId != null) {
            DocHelper.deleteScroll(hostPort, Arrays.asList(lastScrollId));
        }
    }

    public static void migrate(RestHighLevelClient client,String formIndex,String toIndex) throws IOException {
        SearchRequest searchRequest = new SearchRequest(formIndex);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        Scroll scroll = new Scroll(TimeValue.timeValueHours(5L));
        searchSourceBuilder.size(10000);
        searchSourceBuilder.sort("_doc");
        searchRequest.source(searchSourceBuilder);
        searchRequest.scroll(scroll);

        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        String scrollId = searchResponse.getScrollId();
        SearchHits hits = searchResponse.getHits();
        SearchHit[] searchHits = hits.getHits();
        while (searchHits != null && searchHits.length > 0) {
            BulkRequest bulkRequest = new BulkRequest();
            bulkRequest.timeout("5m");
            bulkRequest.setRefreshPolicy("wait_for");
            for(SearchHit searchHit : searchHits){
                IndexRequest indexRequest = new IndexRequest(toIndex).id(searchHit.getId());
                indexRequest.source(searchHit.getSourceAsString(), XContentType.JSON);
                bulkRequest.add(indexRequest);
            }
            client.bulk(bulkRequest, RequestOptions.DEFAULT);
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = searchResponse.getScrollId();
            searchHits = searchResponse.getHits().getHits();
        }
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest(); //完成滚动后，清除滚动上下文
        clearScrollRequest.addScrollId(scrollId);
        ClearScrollResponse clearScrollResponse = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        clearScrollResponse.isSucceeded();
    }
}
